package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.constant.MqttTopicConst;
import cn.tedu.charging.common.pojo.message.CheckResultMessage;
import cn.tedu.charging.common.pojo.message.ProgressMessage;
import cn.tedu.charging.order.service.ConsumerService;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class MqttClientCallback implements MqttCallbackExtended {
    /*
        方法作用: 当mqttClient调用connect并且成功和emqx建立通信之后立即调用
        参数:
        boolean reconnect: 如果本次建立的连接是重新连接,reconnect就是true 第一次连接false
        String serverURI: 连接服务端address tcp://192.168.8.100:1883
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if (reconnect){
            log.info("重新连接成功,serverURI={}",serverURI);
        }else{
            log.info("首次连接成功,serverURI={}",serverURI);
        }
    }
    /*
        方法作用: 当mqttClient调用disconnect方法断开连接后,或者由于网络波动,
        导致和服务端连接通信超时失败,回调就会调用这个方法
        参数:
        Throwable cause: 断开连接的原因
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.info("连接丢失,原因={}",cause.getMessage());
    }
    /*
      方法作用: 当mqttClient发送消息的时候,只要消息成功发送到emqx,不管emqx是否能够接收到
      反馈接收成功还是失败,都会调用这个deliveryComplete,处理发送完成的业务
      qos服务质量级别中0 表示根本不关心投递发送结果,所以,如果客户端mqttClient使用qos=0级别发送消息,不会有任何异常
      不会回调这个方法
      参数:
      IMqttDeliveryToken token: 发送的消息对象 映射,包含了你发送消息的所有内容,以及双方通信的结果
    */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        //1.解析一下 mqttDeliveryToken参数 ack(true表示emqx接收成功 false失败)
        boolean ack = token.isComplete();
        //2.在从token拿到一个异常,如果服务端处理接收的消息失败.就会反馈异常
        MqttException exception = token.getException();
        //3.两个参数同时判断 才能得到我们 是否发送消息真的成功
        if(ack&&exception==null){
            log.info("消息发送成功");
        }else if (exception!=null){
            log.info("emqx接收到消息,但是处理异常失败:{}",exception.getMessage());
        }else if(!ack){
            log.info("消息发送到达emqx失败");
        }

    }
    /**
     * 消费消息的入口 所有当前客户端监听订阅的主题消息都会进入这个方法处理
     * 方法作用: 当mqttClient连接对象订阅了某个主题,一旦这个主题中出现消息,
     * 就会通过这个方法将消息传递给客户端
     * 参数:
     * String topic: 消息来源的topic名称
     * MqttMessage message: 消息对象,包含了消息的所有内容
     */
    @Autowired
    private ConsumerService consumerService;
    @Autowired
    private RedisTemplate redisTemplate;
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        /*log.info("收到消息,topic={},message={}",topic,message.toString());*/
        //1.判断主题来源 目前来源就是自检结果
        if(topic.equals(MqttTopicConst.GUN_CHECK_RESULT_TOPIC)){
            log.debug("收到自检结果消息,topic={},message={}",topic,message.toString());
            CheckResultMessage msg = JSON.parseObject(message.toString(), CheckResultMessage.class);
            //添加抢锁逻辑
            ValueOperations valueOps = redisTemplate.opsForValue();
            //1.设计一个锁的key值,和订单有关
            String lockKey = "charging:order:consume:lock:" + msg.getOrderNo();
            //2.使用do while循环抢锁
            Boolean haveLock = false;//抢锁结果
            int count = 0;//重抢次数的计数器
            try{
                do{
                    //所有线程进入到do都要先抢锁 setnx key value ex 5
                    if (count == 0) {
                        log.debug("第一次抢锁,直接抢");
                        haveLock = valueOps.setIfAbsent(lockKey, "", 5, TimeUnit.SECONDS);
                    } else if (count < 4) {
                        log.debug("第{}次抢锁,失败,正在重试", count + 1);
                        Thread.sleep(200);
                        haveLock = valueOps.setIfAbsent(lockKey, "", 5, TimeUnit.SECONDS);
                    } else {
                        log.error("已经抢锁4次以上,不允许重抢");
                        break;
                    }
                    //计数器+1
                    count++;
                }while (!haveLock);
                log.debug("抢锁成功,开始处理自检结果");
                consumerService.handleCheckResult(msg);
            }catch (Exception e){
                log.error("处理自检结果异常");
            }finally {
                //5.释放锁
                redisTemplate.delete(lockKey);
            }
        }else if(topic!=null&&topic.equals(MqttTopicConst.CHARGING_PROGRESS_TOPIC)){
            //1.将消息解析成ProgressMessage
            ProgressMessage progressMessage = JSON.parseObject(message.toString(), ProgressMessage.class);
            //2.调用消费业务完成消费逻辑
            consumerService.handleChargingProgress(progressMessage);
        }
    }
}
